package com.leyantech.ecp.agenx.dispatcher.di;

import com.leyantech.chaos.kafka.KafkaConsumerStringClient;
import com.leyantech.chaos.kafka.KafkaProducerStringClient;
import com.leyantech.ecp.agenx.common.sender.KafkaSender;
import com.leyantech.ecp.agenx.dispatcher.codec.ClientMessageDecoder;
import com.leyantech.ecp.agenx.dispatcher.codec.ClientMessageEncoder;
import com.leyantech.ecp.agenx.dispatcher.codec.PlainMessageDecoder;
import com.leyantech.ecp.agenx.dispatcher.codec.PlainMessageEncoder;
import com.leyantech.ecp.agenx.dispatcher.consumer.InPutMessageConsumer;
import com.leyantech.ecp.agenx.dispatcher.consumer.OutPutMessageConsumer;

import com.google.inject.AbstractModule;
import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.Provides;
import com.google.inject.Singleton;
import com.google.inject.name.Named;
import io.vertx.core.impl.VertxInternal;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-11-17.
 */
public class DispatcherModule extends AbstractModule {

  @Override
  protected void configure() {

    bind(ClientMessageDecoder.class).to(PlainMessageDecoder.class).asEagerSingleton();

    bind(ClientMessageEncoder.class).to(PlainMessageEncoder.class).asEagerSingleton();

    bind(KafkaSender.class)
        .toProvider(new Provider<KafkaSender>() {

          @Inject
          VertxInternal vertx;

          @Inject
          KafkaProducerStringClient producer;

          @Override
          public KafkaSender get() {
            return new KafkaSender(vertx, producer, "client.message.topic");
          }

        })
        .asEagerSingleton();
  }


  @Singleton
  @Provides
  @Named("output")
  public KafkaConsumerStringClient outputMessageConsumerProvider(
      OutPutMessageConsumer outPutMessageConsumer) {
    return KafkaConsumerStringClient.newBuilder("output-dispatch")
        .setProcessor(outPutMessageConsumer)
        .build();
  }

  @Singleton
  @Provides
  @Named("input")
  public KafkaConsumerStringClient inputMessageConsumerProvider(
      InPutMessageConsumer inPutMessageConsumer) {
    return KafkaConsumerStringClient.newBuilder("input-dispatch")
        .setProcessor(inPutMessageConsumer)
        .build();
  }

  @Singleton
  @Provides
  public KafkaProducerStringClient kafkaProducer() {
    return KafkaProducerStringClient.newBuilder("input-dispatch").build();
  }

}
